package com.runjian.rundo.iot.agent.common.network.tcp.client;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.function.BiConsumer;

import com.runjian.rundo.iot.agent.common.codec.DeviceMessageCodec;
import com.runjian.rundo.iot.agent.common.network.core.BaseNetworkClient;
import com.runjian.rundo.iot.agent.common.network.core.NetworkClient;
import com.runjian.rundo.iot.agent.common.network.tcp.client.config.TcpNetworkClientProperties;
import com.runjian.rundo.iot.agent.common.network.tcp.client.handle.Iec104MessageDecoder;
import com.runjian.rundo.iot.agent.common.network.tcp.client.handle.TcpClientMessageHandler;
import com.runjian.rundo.iot.agent.common.network.tcp.client.session.TcpClientDeviceSessionManager;
import com.runjian.rundo.iot.agent.common.network.tcp.config.ChannelAttributes;
import com.runjian.rundo.iot.agent.common.network.tcp.handle.TcpIdleEventHandler;
import com.runjian.rundo.iot.agent.common.network.tcp.parser.PayloadParser;
import com.runjian.rundo.iot.agent.common.thread.CommonThreadFactory;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * Tcp netty客户端实现
 *
 * @author CaoQingHua
 */
@Slf4j
@Getter
@Setter
@RequiredArgsConstructor
public class TcpNetworkClient extends BaseNetworkClient<TcpNetworkClientProperties> implements NetworkClient<TcpNetworkClientProperties> {

    private final TcpNetworkClientProperties clientProperties;

    protected final TcpClientDeviceSessionManager sessionManager;

    private final TcpIdleEventHandler idleEventHandler;

    private final DeviceMessageCodec codec;

    private final PayloadParser payloadParser;

    private final TcpClientMessageHandler messageHandler;

    /**
     * 循环事件组
     */
    private EventLoopGroup eventLoopGroup;

    /**
     * 连接管道
     */
    private Channel channel;

    /**
     * bootstrap
     */
    Bootstrap bootstrap;

    /**
     * 初始化,只调用一次,后续不再更新
     */
    public void initial() {
        LoggingHandler loggingHandler = new LoggingHandler(clientProperties.getLogLevel());
        // 事件循环组
        eventLoopGroup = new NioEventLoopGroup(new CommonThreadFactory(
            StrUtil.format("NettyClientGroup-{}-{}", this.clientProperties.getHost(), this.clientProperties.getPort())));
        this.bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
            // 设置客户端的NIO线程组
            .channel(NioSocketChannel.class)
            .remoteAddress(new InetSocketAddress(this.clientProperties.getHost(), this.clientProperties.getPort()))
            // 设置是否开启NIO的优化
            .option(ChannelOption.SO_KEEPALIVE, true)
            // 设置是否开启Nagle算法，true表示关闭，false表示开启，通俗地说，如果要求高实时性，有数据发送时就马上发送，就设置为true，如果需要减少发送次数减少网络交互，就设置为false
            .option(ChannelOption.TCP_NODELAY, true)
            // 设置是否开启TCP底层心跳机制
            .option(ChannelOption.SO_REUSEADDR, true)
            // 连接超时毫秒数
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientProperties.getConnectionTimeoutSeconds() * 1000)
            // 添加初始化处理器
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.attr(ChannelAttributes.CONNECTION_ID).set(clientProperties.getId());
                    // 获取管道
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast(loggingHandler);
                    // 添加空闲处理器,每个连接必须维一
                    pipeline.addLast(
                        new IdleStateHandler(clientProperties.getReaderIdleTimeSeconds(), clientProperties.getWriterIdleTimeSeconds(),
                            clientProperties.getReaderIdleTimeSeconds() + clientProperties.getWriterIdleTimeSeconds()));
                    // 添加空闲处理器
                    pipeline.addLast(idleEventHandler);
                    //添加拆包器
                    pipeline.addLast(new Iec104MessageDecoder());
                    // 添加消息处理器
                    if (messageHandler != null) {
                        pipeline.addLast(messageHandler);
                    }
                }
            });
    }

    /**
     * add message decoder
     *
     * @param pipeline the channel pipeline
     */
    protected void addMessageDecoder(ChannelPipeline pipeline) {
        if (payloadParser.getParser() != null) {
            pipeline.addLast(payloadParser.getParser());
        }
    }

    /**
     * 连接成功后,调用异步事件涵数,没用调用就意味着未清空属性
     * 
     * @param callbacks 回调方法
     * @return 返回处理程序
     */
    @Override
    public Future<Void> channelConnection(BiConsumer<Boolean, Channel>... callbacks) {
        ChannelFuture future = null;
        try {
            // 发起异步连接操作，同步阻等待结果
            future = bootstrap.connect().sync();
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        channel = channelFuture.channel();
                        log.info("tcp客户端连接成功,连接信息:{}", clientProperties);
                    } else {
                        log.info("tcp客户端连接失败,连接信息:{}", clientProperties);
                    }
                    if (ObjectUtil.isNotEmpty(callbacks)) {
                        Arrays.stream(callbacks).forEach(callback -> {
                            callback.accept(channelFuture.isSuccess(), channelFuture.channel());
                        });
                    }
                }
            });
            if (!future.isSuccess()) {
                eventLoopGroup.shutdownGracefully();
            }
            //// 等待客户端链路关闭
            // channel().closeFuture().sync();
            // 释放NIO线程组
            // eventLoopGroup.shutdownGracefully();
        } catch (Exception e) {
            log.error("TCP客户端连接失败,异常信息:", e);
            if (ObjectUtil.isNotEmpty(callbacks)) {
                Arrays.stream(callbacks).forEach(callback -> {
                    callback.accept(Boolean.FALSE, null);
                });
            }
        }
        return future;
    }

    /**
     * 停止方法
     */
    @Override
    public void stop() {
        try {
            if (channel != null) {
                this.channel.closeFuture().sync();
            }
        } catch (Exception e) {
            log.error("Tcp连接停止异常");
        } finally {
            // 释放NIO线程组
            eventLoopGroup.shutdownGracefully();
        }
    }

    @Override
    public String getId() {
        return this.clientProperties.getId();
    }

    @Override
    public Boolean status() {
        if (this.channel != null) {
            return this.channel.isActive();
        } else {
            return Boolean.FALSE;
        }
    }
}
